#ifndef M_SERVE_H
#define M_SERVE_H
#include <unordered_map>
#include <vector>
#include <string.h>
#include <vector>
#include <cstdint>
#include <unistd.h>
#include <algorithm>
#include <cassert>
#include <cstring>
#include <iostream>
#include <unistd.h>
#include <sys/types.h>
#include <sys/socket.h>
#include <netinet/in.h>
#include <arpa/inet.h>
#include <fcntl.h>
#include <bits/types.h>
#include <functional>
#include <sys/epoll.h>
#include <time.h>
#include <string>
#include <mutex>
#include <thread>
#include <memory>
#include <sys/eventfd.h>
#include <sys/timerfd.h>
#include <condition_variable>
#include <signal.h>
#define INF 0
#define DBG 1
#define ERR 2
// 在 serve.hpp 中确保日志级别足够详细
#define LOG_LEVEL INF
std::string getlevle(int le)
{
    if (le == 0)
        return "INF";
    else if (le == 1)
        return "DBG";
    else if (le == 2)
        return "ERR";
    else
        return "none";
}
// [可读性很好的时间] [⽇志等级] [进程pid] [打印对应⽇志的⽂件名][⾏号] - 消息内容，⽀持可变参数
// [2024-08-04 12:27:03] [DEBUG] [202938] [main.cc] [16] - hello world
#define LOG(loglevel, format, ...)                                                                                                                             \
                                                                                                                                                               \
    do                                                                                                                                                         \
    {                                                                                                                                                          \
        if (loglevel < LOG_LEVEL)                                                                                                                              \
            break;                                                                                                                                             \
        time_t tim = time(NULL);                                                                                                                               \
        struct tm *ret = localtime(&tim);                                                                                                                      \
        char temp[64] = {0};                                                                                                                                   \
        strftime(temp, 63, "[%Y-%m-%d %H:%M:%S ]", ret);                                                                                                       \
        std::string leve = getlevle(loglevel);                                                                                                                 \
        fprintf(stdout, " %s [%s] [%d] [%lu] [%s] [%d] - " format "\n", temp, leve.c_str(), (int)getpid(), pthread_self(), __FILE__, __LINE__, ##__VA_ARGS__); \
    } while (0)

#define INF_LOG(format, ...) LOG(INF, format, ##__VA_ARGS__)
#define DBG_LOG(format, ...) LOG(DBG, format, ##__VA_ARGS__)
#define ERR_LOG(format, ...) LOG(ERR, format, ##__VA_ARGS__)

#define BUFFER_SIZE 1024

class Buffer
{
public:
    void clear()
    {
        _write_index = 0;
        _read_index = 0;
    }
    Buffer() : _buffer(BUFFER_SIZE), _read_index(0), _write_index(0) {}
    // 提供读取方法 但不移动读指针 从缓冲区读取len个字节到我们得buffer
    void Read(void *buffer, uint64_t len)
    {

        // 下面就是拷贝 因此我们要知道vector读写位置的实际地址
        assert(len <= ReadSpace());
        // 下面就是从我们的vector中拷贝我们的数据到buffer中
        std::copy(ReadPOS(), ReadPOS() + len, (char *)buffer);
    }
    void ReadAndPop(void *buffer, uint64_t len)
    {
        Read(buffer, len);
        MoveReadInd(len);
    }
    std::string ReadAsString(uint64_t len)
    {
        // 要求要获取的数据大小必须小于可读数据大小
        assert(len <= ReadSpace());
        std::string str;
        str.resize(len);
        Read(&str[0], len);
        return str;
    }
    std::string ReadAsStringAndPop(uint64_t len)
    {
        std::string str = ReadAsString(len);
        MoveReadInd(len);
        return str;
    }

    void Write(const void *data, uint64_t len)
    {
        // 写之前确保我们得可用容量足够  ---就是先看总的空间是否满足不满足就按需进行扩容
        if (len <= 0)
            return;
        EnsureSpaceEnough(len);
        // 拷贝data数据加入我们的vector中
        const char *d = (const char *)data;
        // DBG_LOG("copy(d, d + len, WritePos()),len=%ld %s", len, d);
        std::copy(d, d + len, WritePos());
    }
    void WriteAndPop(const void *data, uint64_t len)
    {
        Write(data, len);
        MoveWriteInx(len);
    }
    void WriteString(const std::string &data)
    {
        return Write(data.c_str(), data.size());
    }
    void WriteStringAndPush(const std::string &data)
    {
        WriteString(data);
        MoveWriteInx(data.size());
    }

    void WriteBuffer(Buffer &data)
    {
        return Write(data.ReadPOS(), data.ReadSpace());
    }
    void WriteBufferAndPush(Buffer &data)
    {
        WriteBuffer(data);
        MoveWriteInx(data.ReadSpace());
    }
    // 每次读取我们的\n之前到\n的数据 但不移动读指针
    // C 库函数 void *memchr(const void *str, int c, size_t n)
    // 在参数 str 所指向的字符串的前 n 个字节中搜索第一次出现字符 c（一个无符号字符）的位置。
    std::string GetLine()
    {
        // 找到可读的起始位置
        char *ret = (char *)std::memchr(ReadPOS(), '\n', ReadSpace());
        if (ret == NULL)
            return "";
        else
        {
            // 此时我们的ret就是指向我们'\n'
            //+1把换行符拿到
            return ReadAsString(ret - ReadPOS() + 1);
        }
    }
    std::string GetLineAndPop()
    {
        // ret里面如果读取成功就是带有\n的
        std::string ret = GetLine();
        MoveReadInd(ret.size());
        return ret;
    }

    // private:
    uint64_t ReadSpace() { return _write_index - _read_index; }
    uint64_t HeadSpace() { return _read_index; }
    uint64_t TailSpace() { return _buffer.size() - _write_index; }
    uint64_t WriteSpace() { return HeadSpace() + TailSpace(); }
    char *Begin() { return &(*(_buffer.begin())); }
    char *ReadPOS() { return Begin() + _read_index; }
    char *WritePos() { return Begin() + _write_index; }
    void MoveReadInd(uint64_t len)
    {
        assert(len <= ReadSpace());
        _read_index += len;
    }
    void MoveWriteInx(uint64_t len)
    {
        assert(len <= TailSpace());
        _write_index += len;
    }
    void EnsureSpaceEnough(uint64_t len)
    {
        // 判断len
        if (len <= TailSpace())
            return;
        if (len <= WriteSpace())
        {
            uint64_t readspace = ReadSpace();
            // 此时我们把数据进行拷贝到开始
            std::copy(ReadPOS(), ReadPOS() + readspace, Begin());
            // 重置我们的读写偏移
            _read_index = 0;
            _write_index = readspace;
        }
        else
        {
            // 此时我们只要空间扩容即可 不改变我们的读写偏移
            _buffer.resize(_write_index + len);
        }
    }
    // 我们的读写缓冲区
    std::vector<char> _buffer;
    // 读位置
    uint64_t _read_index;
    // 写位置
    uint64_t _write_index;
};

#define backlogsize 100
class Socket
{
public:
    Socket() : _sockfd(-1) {}
    Socket(int fd) : _sockfd(fd) {}
    ~Socket() { Close(); }

    int Accept()
    {
        // 获取新连接
        struct sockaddr_in peer;
        socklen_t len = sizeof(peer);
        int ret = accept(_sockfd, (sockaddr *)&peer, &len);
        if (ret < 0)
        {
            ERR_LOG("accept error");
            return -1;
        }
        return ret;
    }
    int Fd()
    {
        return _sockfd;
    }
    // send的flag为0就是代表正常发送数据
    // 如果需要发送带外数据啥的可以设置
    // flag为0有阻塞和非阻塞
    int Send(const void *data, int len, int flag = 0)
    {
        if (len == 0)
            return 0;
        // 发送数据
        ssize_t ret = send(_sockfd, data, len, flag);
        // 此时发送失败返回要考虑非阻塞的情况
        if (ret < 0)
        {
            // 此时看看是不是被信号中断了
            if (errno == EAGAIN || errno == EINTR)
            {
                return 0;
            }
            ERR_LOG("send error");
            return -1;
        }
        return ret;
    }
    int SendNoblock(void *data, int len)
    {
        return Send(data, len, MSG_DONTWAIT);
    }
    int Recv(void *buffer, int len, int flag = 0)
    {
        // DBG_LOG(" Recv(void *buffer, int len, int flag = 0);len=%d", len);
        if (len == 0)
        {
            return 0;
        }
        ssize_t ret = recv(_sockfd, buffer, len, flag);
        int ret1 = ret;
        if (ret < 0)
        {
            // DBG_LOG(" Recv(void *buffer, int len, int flag = 0);读取长度=%d", ret1);
            if (errno == EAGAIN || errno == EINTR)
            {
                // DBG_LOG(" Recv(void *buffer, int len, int flag = 0);信号打断读取，读取长度=%d", ret1);
                return 0;
            }
            // 这里面出错的情况包括了我们的fd已经被关闭了
            ERR_LOG("recv error");
            return -1;
        }
        else
        {
            // 注意走到这里有可能是对端关闭连接了 我们这里返回错误
            if (ret == 0)
            {
                DBG_LOG("对端关闭连接了");
                return -1;
            }
            // DBG_LOG(" Recv(void *buffer, int len, int flag = 0);读取长度=%d", ret1);
            return ret;
        }
    }
    int RecvNoblock(void *buffer, int len)
    {
        return Recv(buffer, len, MSG_DONTWAIT);
    }
    // 默认是可以从对端的任意ip连接
    bool CreateServe(uint16_t port, const std::string &ip = "0.0.0.0", bool block_flag = false)
    {
        // 1.进行我们的创建套接字
        bool ret = Create();
        if (ret == false)
        {
            ERR_LOG("serve create error:socket error");
            return false;
        }
        // 设置套接字地址进行复用
        ret = SetPortReuse();
        if (ret == false)
        {
            ERR_LOG("serve create error:SetPortReuse error");
            return false;
        }
        if (block_flag)
            // 设置非阻塞
            SetNoblock();

        // 绑定端口号
        ret = Bind(ip, port);
        if (ret == false)
        {
            ERR_LOG("serve create error:bind error");
            return false;
        }
        // 进行监听
        ret = Listen();
        if (ret == false)
        {
            ERR_LOG("serve create error:listen error");
            return false;
        }
        return true;
    }
    bool CreateClient(std::string ip, uint16_t port)
    {
        // 创建套接字
        bool ret = Create();
        if (ret == false)
        {
            ERR_LOG("client error:socket error");
            return false;
        }
        // 进行connect
        ret = Connect(ip, port);
        if (ret == false)
        {
            ERR_LOG("serve create error:connect error");
            return false;
        }
        return true;
    }
    void Close()
    {
        if (_sockfd != -1)
            close(_sockfd);
    }

private:
    // 创建套接字
    bool Create()
    {
        int n = socket(AF_INET, SOCK_STREAM, IPPROTO_TCP);
        if (n < 0)
        {
            ERR_LOG("socket error");
            return false;
        }
        _sockfd = n;
        return true;
    }
    // bind端口
    bool Bind(std::string ip, uint16_t port)
    {
        // int bind(int sockfd, const struct sockaddr *addr, socklen_t addrlen);
        struct sockaddr_in localhost;
        localhost.sin_family = AF_INET;
        localhost.sin_port = htons(port);
        //  int inet_pton(int af, const char *src, void *dst);
        int ret = inet_pton(AF_INET, ip.c_str(), &localhost.sin_addr);
        if (ret < 0)
        {
            ERR_LOG("inet_pton erroe");
            return false;
        }
        ret = bind(_sockfd, (sockaddr *)&localhost, sizeof(localhost));
        if (ret < 0)
        {
            ERR_LOG("Bind Error");
            return false;
        }
        return true;
    }
    // 监听状态
    bool Listen(int backlog = backlogsize)
    {
        // int listen(int sockfd, int backlog);
        int ret = listen(_sockfd, backlog);
        if (ret < 0)
        {
            ERR_LOG("listen erroe");
            return false;
        }
        return true;
    }
    // connect连接
    bool Connect(std::string ip, uint16_t port)
    {
        // int connect(int sockfd, const struct sockaddr *addr,socklen_t addrlen);
        sockaddr_in peer;
        peer.sin_family = AF_INET;
        peer.sin_port = htons(port);
        int ret = inet_pton(AF_INET, ip.c_str(), &peer.sin_addr);
        if (ret < 0)
        {
            ERR_LOG("inet_pton error");
            return false;
        }
        ret = connect(_sockfd, (sockaddr *)&peer, sizeof(peer));
        if (ret < 0)
        {
            ERR_LOG("connect error");
            return false;
        }
        return true;
    }
    // 设置地址 端口复用
    bool SetPortReuse()
    {
        // int setsockopt(int sockfd, int level, int optname,const void *optval, socklen_t optlen);
        int val = 1;
        int ret = setsockopt(_sockfd, SOL_SOCKET, SO_REUSEADDR, &val, sizeof(int));
        if (ret < 0)
        {
            ERR_LOG("setsockopt reuseaddr error");
            return false;
        }
        val = 1;
        ret = setsockopt(_sockfd, SOL_SOCKET, SO_REUSEPORT, &val, sizeof(int));
        if (ret < 0)
        {
            ERR_LOG("setsockopt reuseport error");
            return false;
        }
        return true;
    }
    // 设置非阻塞读取
    void SetNoblock()
    {
        //   int fcntl(int fd, int cmd, ... /* arg */ );
        int flag = fcntl(_sockfd, F_GETFL, 0);
        fcntl(_sockfd, F_SETFL, flag | O_NONBLOCK);
    }

private:
    int _sockfd;
};

// 改模块负责一个文件描述符的事件管理 和事件处理
class poller;
class EventLoop;
class channel
{
public:
    using callback = std::function<void()>;
    channel(int fd, EventLoop *poll)
        : _fd(fd), _event(0), _revent(0),
          _write_callback(nullptr), _read_callback(nullptr), _errro_callback(nullptr), _close_callback(nullptr), _allevent_callback(nullptr), _poll(poll)
    {
    }
    uint32_t GetEvent()
    {
        return _event;
    }
    // 开启对读事件的关心 并写入内核中
    void EnableRead()
    {
        if (!IsReadEvent())
        {
            // 开启对读事件的关心
            _event |= EPOLLIN;
        }
        // TODO 写入内核
        return IntoEpoll();
    }

    // 开启对写事件的关心 并写入内核中
    void EnableWrite()
    {
        if (!IsWriteEvent())
        {
            _event |= EPOLLOUT;
        }
        // todo 写入内核
        return IntoEpoll();
    }
    // 关闭对读事件的关心 并从内核从移除对改文件描述符的读事件的关心
    void DisableRead()
    {
        if (IsReadEvent())
        {
            _event &= ~EPOLLIN;
        }
        // toDO 内核操作
        return IntoEpoll();
    }
    // 关闭对写事件的关心 并从内核从移除对改文件描述符的写事件的关心
    void DisableWrite()
    {
        if (IsWriteEvent())
        {
            _event &= ~EPOLLOUT;
        }
        // todo 对epoll实例操作
        return IntoEpoll();
    }
    // 关闭所有事件的关心
    void DisableAllEvent()
    {
        _event = 0;
        // 设置到内核中
        return IntoEpoll();
    }
    // 从epoll中移除对fd的管理
    void Remove();

    // 经过一次轮询之后设置就绪的事件
    void SetRevent(uint32_t event)
    {
        _revent = event;
    }
    // 判断是否关心读事件
    bool IsReadEvent()
    {
        return _event & EPOLLIN;
    }
    // 判断是否关心写事件
    bool IsWriteEvent()
    {
        return _event & EPOLLOUT;
    }
    // 根据就绪的事件进行函数的执行
    void HandlerEvents(int timerfd)
    {
        // DBG_LOG("timerfd:%d 触发事件fd:%d", timerfd, _fd);
        //  触发了读事件
        if ((_revent & EPOLLIN) || (_revent & EPOLLRDHUP) || (_revent & EPOLLPRI))
        {
            // DBG_LOG("fd:%d的读事件触发了", _fd);
            //  首先执行我们任意事件触发都要执行的函数 执行我们的allevent要确保此时的fd没有被关闭
            // 不用在此时先执行我们的任意事件到来了 因为在处理读事件任务 就算出错了 我们也不会立马关闭连接而是在我们的
            // 一轮事件处理完之后关闭连接的 此时我们对本轮就绪事件对任意事件操作就不会崩溃的可能了
            // if (_allevent_callback)
            //     _allevent_callback();
            // 每次事件的回调函数一定确保在连接释放前 进行调用
            if (_allevent_callback && _fd != timerfd)
                _allevent_callback();
            if (_read_callback)
                _read_callback();
        }
        // 如果写事件之后关闭了文件描述符 下面我们接着多任意事件的触发就会报错
        if (_revent & EPOLLOUT)
        {
            if (_allevent_callback && _fd != timerfd)
                _allevent_callback();
            // DBG_LOG("fd:%d的写事件触发了", _fd);
            if (_write_callback)
                _write_callback();
        }
        else if (_revent & EPOLLERR)
        {
            // DBG_LOG("fd:%d的错误事件触发了", _fd);
            //  此时关闭文件符 后续的话就算还有其他事件就绪 也不需要执行了
            if (_allevent_callback && _fd != timerfd)
                _allevent_callback();
            if (_errro_callback)
                _errro_callback();
        }
        else if (_revent & EPOLLHUP)
        {
            if (_allevent_callback && _fd != timerfd)
                _allevent_callback();
            // DBG_LOG("fd:%d的连接断开事件触发了", _fd);
            if (_close_callback)
                _close_callback();
        }
    }
    // 获取文件描述符
    int Fd()
    {
        return _fd;
    }
    void SetReadCallback(callback cb) { _read_callback = cb; }
    void SetWriteCallback(callback cb) { _write_callback = cb; }
    void SetAllEventCallback(callback cb) { _allevent_callback = cb; }
    void SetCloseCallback(callback cb) { _close_callback = cb; }
    void SetErrorCallback(callback cb) { _errro_callback = cb; }
    // 把事件设置到内核中
    void IntoEpoll();

private:
    int _fd;
    uint32_t _event;
    uint32_t _revent;
    EventLoop *_poll;
    callback _read_callback;
    callback _write_callback;
    callback _allevent_callback;
    callback _close_callback;
    callback _errro_callback;
};

#define EPOLLEVENTSIZE 1024

class poller
{
public:
    poller()
    {
        _epfd = epoll_create(EPOLLEVENTSIZE);
        if (_epfd < 0)
        {
            ERR_LOG("epoll_create error");
            abort(); // 退出程序
        }
    }
    // 把一个文件描述符关心的事件写如epoll内核中
    // 并加入我的管理中
    void UpdateEvent(channel *ch)
    {
        if (IsInEvents(ch))
        {
            // 更新操作
            return upadate(ch, EPOLL_CTL_MOD);
        }
        _events.insert(std::make_pair(ch->Fd(), ch));
        // 添加操作
        return upadate(ch, EPOLL_CTL_ADD);
    }
    // 从epoll中移除某个文件描述符的管理
    void RemoveEvent(channel *ch)
    {
        auto it = _events.find(ch->Fd());
        if (it != _events.end())
        {
            _events.erase(ch->Fd());
        }
        upadate(ch, EPOLL_CTL_DEL);
    }
    // 单次的监控一次loop
    void OnceLoop(std::vector<channel *> &active)
    {
        int n = epoll_wait(_epfd, _revents, EPOLLEVENTSIZE, -1);
        if (n < 0)
        {
            if (errno == EINTR) // 被信号中断
            {
                return;
            }
            ERR_LOG("EPOLL WAIT ERROR:%s\n", strerror(errno));
            abort(); // 退出程序
        }
        for (int i = 0; i < n; i++)
        {
            auto ch = _revents[i];
            int fd = ch.data.fd;
            auto tem = _events.find(fd);
            if (tem == _events.end())
            {
                ERR_LOG("RemoveEvent Error");
                abort();
            }
            channel *chan = tem->second;
            chan->SetRevent(ch.events);
            active.push_back(chan);
        }
        return;
    }

private:
    bool IsInEvents(channel *ch)
    {
        int fd = ch->Fd();
        auto it = _events.find(fd);
        if (it == _events.end())
        {
            return false;
        }
        return true;
    }
    void upadate(channel *ch, int op)
    {
        int fd = ch->Fd();
        struct epoll_event event;
        event.data.fd = fd;
        event.events = ch->GetEvent();

        int n = epoll_ctl(_epfd, op, fd, &event);
        if (n < 0)
        {
            if (errno == EBADF)
            {
                DBG_LOG("epoll_ctl 忽略 Bad file descriptor: op=%d, fd=%d", op, fd);
                return;
            }
            ERR_LOG("epoll_ctl error: op=%d, fd=%d, events=%u, errno=%d (%s)",
                    op, fd, event.events, errno, strerror(errno));
            // ERR_LOG("epoll_ctl error");
            // abort();
            return;
        }
        return;
    }

private:
    int _epfd;
    struct epoll_event _revents[EPOLLEVENTSIZE];
    std::unordered_map<int, channel *> _events;
};

using TaskFunc = std::function<void()>;
using RemoveFunc = std::function<void()>;
// 一个定时任务
class TimerTask
{
public:
    TimerTask(int id, int timeout, TaskFunc func) : _id(id), _taskfunc(func), _timeout(timeout), _cancel(false) {}
    ~TimerTask()
    {
        // 当我们的定时任务生命周期结束的时候执行我们的任务函数
        if (_cancel == false)
            _taskfunc();
        // 当我们的定时任务被执行后 我们要移除我们的定时器在容器中的管理

        // 无论定时任务是否被取消 在定时任务析构的时候我们都要从容器中移除对改定时任务的管理
        _remove(); // 移除我们在容器中的管理
    }
    void Cancel()
    {
        _cancel = true;
    }
    void RegisterRemoveFunc(RemoveFunc func)
    {
        _remove = func;
    }
    int GetTimeout()
    {
        return _timeout;
    }
    int _id; // 每个定时任务的编号

private:
    TaskFunc _taskfunc; // 要执行的我们的任务函数
    RemoveFunc _remove;
    int _timeout; // 超时时间
    bool _cancel; // 定时任务是否被取消
};

class EventLoop;
using TaskSharedPtr = std::shared_ptr<TimerTask>;
using TaskWeakptr = std::weak_ptr<TimerTask>;
class TimerWheel
{
private:
    int _timerfd; // 用于让我们的事件轮每秒钟执行一次函数移动指针从而执行我的定时任务
    // 每个定时器绑定到一个eventllop这样的话设置定时任务就是在一个线程中执行
    // 因为在向定时器任务添加定时任务的时候是线程不安全的 通过loop可以把关于连接的定时任务
    // 安全的放到对应的线程中执行
    EventLoop *_loop; // 每个定时器绑定一个epoll线程
    std::unique_ptr<channel> _timerfd_channel;
    int _tick;                                       // 我们的秒级指针 指向哪里就执行
    int _capacity;                                   // 轮子容量
    std::vector<std::vector<TaskSharedPtr>> _wheels; // 我们的时间轮子
    // 我们的定时时间管理容器 这里用我们的weak是为了防止定时任务的计数最小就是1导致无法触发定时任务的析构函数
    std::unordered_map<int, TaskWeakptr> _timers;

public:
    TimerWheel(EventLoop *loop)
        : _timerfd(CreateTimerfd()),
          _loop(loop),
          _timerfd_channel(new channel(_timerfd, _loop)),
          _tick(0),
          _capacity(60),
          _wheels(_capacity)
    {
        _timerfd_channel->SetReadCallback(std::bind(&TimerWheel::HandleTimefdRead, this));
        // 开启对timerfd的读事件监控
        _timerfd_channel->EnableRead();
    }

    // 三个外部接口有可能会被其他线程操作加入定时任务 因此要确保线程安全让其操作在自己的线程中执行
    void TimerTaskAdd(int id, int timeout, TaskFunc func);

    void CancelTask(int id);
    int TimerFd()
    {
        return _timerfd;
    }
    // 刷新定时任务
    void Refresh(int id);

    // 该函数其实也是线程不安全的但是由于改函数只会在我么timerfd事件触发的时候被执行
    // 由于我们的timerfd事件监控 绑定了该模块的epoll 因此事件触发一定是在该线程中执行
    void RunOneOneSecond()
    {
        _tick = (_tick + 1) % _capacity;
        // std::cout << "指针开始移动:" << _tick << std::endl;
        _wheels[_tick].clear(); // 此时如果定时任务对象计数为0 就会执行
    }
    /*这个接口存在线程安全问题--这个接口实际上不能被外界使用者调用，只能在模块内，在对应的EventLoop线程内执行*/
    bool HasTimer(uint64_t id)
    {
        auto it = _timers.find(id);
        if (it == _timers.end())
        {
            return false;
        }
        return true;
    }

private:
    void RefreshInLoop(int id)
    {
        auto pos = _timers.find(id);
        if (pos == _timers.end())
        {
            return;
        }
        // 找到之后取消定时任务
        TaskSharedPtr tem = _timers[id].lock(); // 把观察的资源重新生成一个sharedptr 此时如果没有资源指向了就会返回空
        // 反之观察的资源的引用计数增加 但是这里不影响 因为我们tem的出作用域后就没了 不影响时间轮事件的引用计数
        if (tem)

        {
            // sleep(1);
            //  std::cout << "-----------------------------------------" << std::endl;
            int timeout = tem->GetTimeout();
            // std::cout << "time out is" << timeout << std::endl;
            //  重新设置时间轮
            // std::cout << "刷新之前的tick:" << _tick << " 刷新之后的tick:" << ((_tick + timeout) % _capacity) << std::endl;
            _wheels[(_tick + timeout) % _capacity].push_back(tem);
        }
    }
    void CancelTaskInLoop(int id)
    {
        auto pos = _timers.find(id);
        if (pos == _timers.end())
        {
            return;
        }

        // 找到之后取消定时任务
        TaskSharedPtr tem = _timers[id].lock(); // 把观察的资源重新生成一个sharedptr 此时如果没有资源指向了就会返回空
        // 反之观察的资源的引用计数增加 但是这里不影响 因为我们tem的出作用域后就没了 不影响时间轮事件的引用计数
        if (tem)
        {
            tem->Cancel(); // 取消定时任务
        }
    }
    void TimerTaskAddInLoop(int id, int timeout, TaskFunc func)
    {
        // 构建一个shared_ptr
        TaskSharedPtr task = std::make_shared<TimerTask>(id, timeout, func);
        // 注册回调定时任务执行会自动把自己从容器中移除
        task->RegisterRemoveFunc(std::bind(&TimerWheel::RemoveTimer, this, id));
        int pos = (_tick + timeout) % _capacity;
        // std::cout << "超时任务首次添加,触发tick:" << pos << std::endl;
        _wheels[pos].push_back(task);    // 加入轮子对应位置
        _timers[id] = TaskWeakptr(task); // 变为我们的weak指针不会增加引用计数
        // 这里就是如果我们用sharedptr 有两种情况
        // 1. 我们直接用此时的task加入容器中 此时就会导致task引用加1
        // 这样的话 就算轮子上的shared——ptr都释放了 但是容器中还有一个引用计数导致 定时任务无法执行
        // 2.如果我们用源对象重新搞一个shared——ptr 这样虽然我们的task此时的引用计数没有增加 但是如果task资源释放了
        // 此时我们容器中管理的定时任务为空了 我们去移除的时候会报错
        // weakptr 就相当于我们的一个资源的观察者
    }
    void HandleTimefdRead()
    {
        // DBG_LOG("_tickbegin:%d",_tick);
        int n = ReadTimerfd();
        for (int i = 0; i < n; i++)
        {
            // 移动时间轮的指针
            RunOneOneSecond();
        }
        // DBG_LOG("_tickafter:%d",_tick);
    }
    int ReadTimerfd()
    {
        uint64_t tmp;
        int n = read(_timerfd, &tmp, 8);
        if (n < 0)
        {
            if (errno == EINTR || errno == EAGAIN)
            {
                return -1;
            }
            ERR_LOG("处理定时器的读事件出错了");
            abort();
        }
        return tmp;
    }
    int CreateTimerfd()
    {
        int fd = timerfd_create(CLOCK_MONOTONIC, 0);
        if (fd < 0)
        {
            ERR_LOG("创建定时器的fd出错了");
            abort();
        }
        // int timerfd_settime(int fd, int flags, struct itimerspec *new, struct itimerspec *old);
        struct itimerspec itime;
        itime.it_value.tv_sec = 1;
        itime.it_value.tv_nsec = 0; // 第一次超时时间为1s后
        itime.it_interval.tv_sec = 1;
        itime.it_interval.tv_nsec = 0; // 第一次超时后，每次超时的间隔时
        _timerfd = fd;
        timerfd_settime(_timerfd, 0, &itime, NULL);
        return _timerfd;
    }
    void RemoveTimer(int id)
    {
        auto pos = _timers.find(id);
        if (pos == _timers.end())
        {
            return;
        }
        _timers.erase(id);
    }
};

// 该模块负责我们一个epoll实列的事件管理 以及一个定时器模块
// 改模块绑定一个线程 例如主线程 把到来的新连接 交给了一个任意一个工作线程
// 此时我们不可以在主线程的epoll中直接添加对连接的监控
// 而是在工作线程中进行 因此要加入工作线程的任务队列
// 同时队列要加锁 因为在主线程分配给某个工作线程分配连接的时候工作线程可能正在取出任务

class EventLoop
{
    using Func = std::function<void()>;

private:
    // 改eventpoll被创建的时候创建的线程id
    std::thread::id _thread_id;

    // 每个都要拥有一个epoll实例
    poller _poll;
    // eventfd用于如果改线程阻塞在epoll_wait导致后续任务队列任务无法执行
    int _eventfd;
    // 用只能指针我们就不用显示实现析构函数了
    std::unique_ptr<channel> _eventfd_channel;
    // 队列
    std::vector<Func> _tasks;
    // 锁
    std::mutex _mutex;
    // 我们的定时器
    TimerWheel _timer_wheel;
    // 定时器的fd
    int _timerfd;

private:
    bool IsMyThread()
    {
        return _thread_id == std::this_thread::get_id();
    }
    void Weakup()
    {
        // 就是向eventfd写入数据
        uint32_t data = 1;
        int n = write(_eventfd, &data, 8);
        if (n < 0)
        {
            if (errno == EAGAIN || errno == EINTR)
            {
                return;
            }
            ERR_LOG("写入eventfd失败");
            abort();
        }
        return;
    }
    void RunAllTasks()
    {
        std::vector<Func> tasks;
        {
            std::unique_lock<std::mutex> lock(_mutex);
            _tasks.swap(tasks);
        }
        // 直接执行我的任务
        for (auto &f : tasks)
        {
            f();
        }
    }
    // 外部线程通过加入队列任务交给我这个线程
    void QueueInLoop(const Func &task)
    {
        // 对队列的操作加速防止放入任务的时候
        // 有其他的线程正在去除任务
        {
            std::unique_lock<std::mutex> lock(_mutex);
            _tasks.push_back(task);
        }
        // 唤醒线程防止在事件监控哪里
        Weakup();
    }
    void HandleEventfdRead()
    {
        // 读取八字节的数据
        uint64_t data;
        int n = read(_eventfd, &data, 8);
        if (n < 0)
        {
            if (errno == EAGAIN || errno == EINTR)
            {
                return;
            }
            ERR_LOG("读取eventfd失败");
            abort();
        }
        return;
    }

    int CreateEventFd()
    {
        // int eventfd(unsigned int initval, int flags);
        int n = eventfd(0, EFD_CLOEXEC | EFD_NONBLOCK);
        if (n < 0)
        {
            ERR_LOG("CreateEventFd Error");
            abort();
        }
        _eventfd = n;
        return _eventfd;
    }

public:
    EventLoop()
        : _thread_id(std::this_thread::get_id()),
          _eventfd(CreateEventFd()),
          _eventfd_channel(new channel(_eventfd, this)), // evenfd把自己加入改epoll的实例中
          _timer_wheel(this),                            // 给定时器绑定到我们的epoll线程中
          _timerfd(_timer_wheel.TimerFd())
    {
        _eventfd_channel->SetReadCallback(std::bind(&EventLoop::HandleEventfdRead, this));
        // 开启读事件监控
        _eventfd_channel->EnableRead();
    }
    bool HasTimer(uint64_t id)
    {
        return _timer_wheel.HasTimer(id);
    }
    // 提供定时器添加任务
    void TimerTaskAdd(int id, int timeout, TaskFunc func)
    {
        _timer_wheel.TimerTaskAdd(id, timeout, func);
    }
    void AssertInLoop()
    {
        assert(_thread_id == std::this_thread::get_id());
    }
    // 删除定时任务
    void TimerTaskCancel(int id)
    {
        _timer_wheel.CancelTask(id);
    }
    // 刷新定时任务
    void RefrushTimerTask(int id)
    {
        _timer_wheel.Refresh(id);
    }

    void UpdateEvent(channel *ch)
    {
        _poll.UpdateEvent(ch);
    }
    void RemoveEvent(channel *ch)
    {
        _poll.RemoveEvent(ch);
    }
    // 该线程开启做的任务
    void Start()
    {
        // 注意这里我们要进行判断定时任务的timerfd如果是读事件就绪 我们只触发读事件回调 而不进行任意事件回调
        while (1)
        {
            // 开启事件循环
            std::vector<channel *> active;
            _poll.OnceLoop(active);
            // 进行一轮事件处理
            for (int i = 0; i < active.size(); i++)
            {
                // 没有连接到来如果设置了我们的定时任务此时也会进行事件就绪
                // DBG_LOG("active size:%d",(int)active.size());
                // INF_LOG("线程id:%lu", pthread_self());
                active[i]->HandlerEvents(_timerfd);
            }
            // 去任务队列的任务进行执行
            RunAllTasks();
        }
    }
    // 通过线程改eventloop的线程id和正在执行该函数的线程id进行判断
    // 如果是自己的线程内的任务直接执行
    // 反之比如主线程派发的新连接监控任务 需要我执行的
    // 加入我的队列等待本次事件循环结束后执行
    // 这就是为啥要有队列 因为外部线程先要用工作线程的epoll来进行监控事件
    // 我们的队列就很好的进行了线程间的通信
    // 比如当前执行的线程不是我这个模块初始绑定的线程 因此直接加入我这个epoll初始线程的队列中去
    // 等待时间循环后 有我自己的线程来执行我自己的任务
    void RunInLoop(const Func &task)
    // 也就是说这个函数保证任务被自己模块初始的线程执行 如果来了其他线程加入我的队列
    // 最后还是在我的线程中执行的
    {
        // 外部通过改接口进行任务执行
        if (IsMyThread())
        {
            task();
        }
        else
        {
            QueueInLoop(task);
        }
        return;
    }
};

class Any
{
    class holder
    {
    public:
        virtual ~holder() {};
        virtual const std::type_info &type() = 0;
        virtual holder *clone() = 0;
    };

    template <typename T>
    class placeholder : public holder
    {
    public:
        placeholder(const T &val) : _val(val) {}
        const std::type_info &type() override { return typeid(T); }
        // 这里回调用T拷贝构造
        holder *clone() override { return new placeholder<T>(_val); }
        T _val;
    };

private:
    holder *_ptr;

public:
    Any() : _ptr(nullptr) {}
    Any &swap(Any &other)
    {
        std::swap(_ptr, other._ptr);
        return *this;
    }
    // Any a=9  Any b("csafd")
    template <typename T>
    Any(const T &val) : _ptr(new placeholder<T>(val)) {}
    // 拷贝构造 注意这里深拷贝 如果我们直接把this->_ptr赋值为other._ptr就是浅拷贝
    // 拷贝构造一定要加引用

    Any(const Any &other) { _ptr = other._ptr ? other._ptr->clone() : nullptr; }

    // 拿到保存的数据指针
    template <typename T>
    T *GetVal()
    {
        assert(_ptr->type() == typeid(T));
        return &(((placeholder<T> *)_ptr)->_val);
    }

    template <typename T>
    Any &operator=(const T &val)
    {
        Any(val).swap(*this);
        return *this;
    }
    Any &operator=(const Any &other)
    {
        Any(other).swap(*this);
        return *this; // 这里会发生一次any的拷贝构造 拷贝
    }
    ~Any()
    {
        delete _ptr;
    }
};

class Connection;
using PtrConnection = std::shared_ptr<Connection>;
// DISCONECTED -- 连接关闭状态；   CONNECTING -- 连接建立成功-但是此时没有进行回调函数的设置
// CONNECTED -- 连接建立完成，各种设置已完成，可以通信的状态；  DISCONNECTING -- 待关闭状态
typedef enum
{
    DISCONNECTED,
    CONNECTING,
    CONNECTED,
    DISCONNECTING
} ConnStatu;
// 该模块就是对一个通信连接的集成 向外提供回调函数进行业务处理 内部设置自己时间触发的回调函数
class Connection : public std::enable_shared_from_this<Connection>
{

private:
    // 连接id 定时器任务id
    uint64_t _connection_id;
    // 连接状态
    ConnStatu _connection_status;
    // 连接的通信文件描述符
    int _sock_fd;
    // 连接通过socket进行信息发送接受
    Socket _socket;
    // 读取数据读到缓冲区中
    Buffer _recv_buffer;
    // 经过处理的数据发送到发送缓冲区中
    Buffer _send_buffer;
    // 因为连接要有自己的epoll实例 以及连接需要进行定时任务的处理 因此绑定eventloop模块
    EventLoop *_loop;
    // 连接的事件管理 通过channel模块 在我们该模块构造的时候直接进行设置
    channel _channel;
    // 上下文信息
    Any _context;
    // 是否开启超时连接关闭
    bool _is_enable_realase;

    // 外部设置的互调函数
    /*这四个回调函数，是让服务器模块来设置的（其实服务器模块的处理回调也是组件使用者设置的）*/
    /*换句话说，这几个回调都是组件使用者使用的*/
    using ConnectedCallback = std::function<void(const PtrConnection &)>;         // 把连接传进去为了我们上层可能会进行连接的其他操作
    using MessageCallback = std::function<void(const PtrConnection &, Buffer *)>; // 把信息带给上层业务处理
    using ClosedCallback = std::function<void(const PtrConnection &)>;
    using AnyEventCallback = std::function<void(const PtrConnection &)>;
    ConnectedCallback _connected_callback; // 连接建立完成初始化后进行的操作
    MessageCallback _message_callback;     // 业务处理函数
    ClosedCallback _closed_callback;       // 当连接断开的时候的 操作
    AnyEventCallback _event_callback;      // 任意事件到来的操作 其实就是刷新我我们的活跃度
    /*组件内的连接关闭回调--组件内设置的，因为服务器组件内会把所有的连接管理起来，一旦某个连接要关闭*/
    /*就应该从管理的地方移除掉自己的信息*/
    // 由于我们管理连接使用shared_ptr因此这个回调函数 在我我们的管理容器所在模块进行设置 从容器中移除
    // 引用计数位0释放我们的连接
    ClosedCallback _server_closed_callback;

private:
    // 读事件 常设
    // 写事件 按需设置 发送缓冲区中有数据开启 没有数据关闭
    void HandlerRead()
    {
        if (_connection_status == DISCONNECTED)
            return;
        // DBG_LOG("进入读事件处理");
        // 读取数据到我们的接受缓冲区
        char buffer[65536] = {0};
        int ret = _socket.RecvNoblock(buffer, sizeof(buffer) - 1);
        int tem = ret;
        // DBG_LOG("ret:%d buffer:%s", tem, &buffer[0]);
        //  注意我们实现的返回为0 不代表客户端关闭连接了 只有为-1的时候才是错的
        if (tem < 0)
        {
            DBG_LOG("读取生错误了");
            // 里面有连接已经关闭的情况
            // 此时我们不要立马关闭连接看一下此时的接受缓存区和发送缓冲区是否还有数据没有处理
            // 调用shutdown的函数的时候肯此时的客户端已经关闭连接了
            ShutDownInLoop();
            return;
        }
        // 现在的话是已经没有任何错了 此后连接一定没有被关闭
        if (_connection_status == DISCONNECTED)
            return;
        _recv_buffer.WriteAndPop(buffer, ret);
        // 下面调用上层的回调函数进行处理
        if (_recv_buffer.ReadSpace() > 0)
        {
            if (_message_callback)
                _message_callback(shared_from_this(), &_recv_buffer);
        }
        return;
    }
    void HandlerWrite()
    {
        if (_connection_status == DISCONNECTED)
            return;
        // 写事件触发 此时连接的fd在这一时刻没有被关闭
        // 1.此时发送数据
        size_t ret = _socket.SendNoblock(_send_buffer.ReadPOS(), _send_buffer.ReadSpace());
        if (ret < 0)
        {
            // 此时进行读取就有可能出错了 比如对端已经关闭连接
            // 此时我们已经不可能通过文件描述符进行通信了
            // 此时我们要做的就是把接受缓冲区的数据进行业务处理
            if (_recv_buffer.ReadSpace() > 0)
            {
                if (_message_callback)
                {
                    _message_callback(shared_from_this(), &_recv_buffer);
                }
            }
            // 这里有一个细节就是我们连接的释放操作 一定是在我们的本次事件处理完毕后进行释放的
            return Realse(); // 直接释放连接
        }
        _send_buffer.MoveReadInd(ret);
        // 注意发送数据可能会没有发送完
        if (_send_buffer.ReadSpace() == 0)
        {
            // 发送完毕 关闭写事件
            if (_connection_status != DISCONNECTED)
            {
                _channel.DisableWrite();
            }
            // 此时数据发送完毕 如果还是半关闭状态直接进行连接释放
            if (_connection_status == DISCONNECTING)
            {
                if (_recv_buffer.ReadSpace() > 0)
                {
                    if (_message_callback)
                    {
                        _message_callback(shared_from_this(), &_recv_buffer);
                    }
                }
                Realse();
            }
            return;
        }
    }
    // 注册给我们每个连接的关闭回调
    void HandlerClose()
    {
        if (_connection_status == DISCONNECTED)
            return;
        if (_recv_buffer.ReadSpace() > 0)
        {
            if (_message_callback)
                _message_callback(shared_from_this(), &_recv_buffer);
        }
        return Realse();
    }
    void HandlerErr()
    {
        if (_connection_status == DISCONNECTED)
            return;
        HandlerClose();
    }
    void HanlerEvent()
    {
        if (_connection_status == DISCONNECTED)
            return;
        // 刷新活跃度
        if (_is_enable_realase == true)
        {
            // 如果开启了定时销毁 如果有任意事件 刷新活跃度
            _loop->RefrushTimerTask(_connection_id);
        }
        if (_event_callback)
        {
            _event_callback(shared_from_this());
        }
        return;
    }
    void EnableConReadInLoop()
    {
        // 此时已经处于连接已经获得了
        assert(_connection_status == CONNECTING);
        // 读事件回调 在构造改连接额时候已经设置好了
        // 因此我们只要开启写事件监控
        _channel.EnableRead();
        // 注意我们一定要设置在外部调用函数 不然就会连接已经就绪但是我们的回调函数没有进行设置
        if (_connected_callback)
        {
            _connected_callback(shared_from_this());
        }
        _connection_status = CONNECTED;
        return;
    }
    void CancelInactiveReleaseInLoop()
    {
        // 取消我们的标志
        _is_enable_realase = false;
        if (_loop->HasTimer(_connection_id))
            _loop->TimerTaskCancel(_connection_id);
    }

    void EnableInactiveRelseInLoop(int delay)
    {
        // 设置我的定时任务位开启
        _is_enable_realase = true;
        // 设置定时任务
        if (_loop->HasTimer(_connection_id))
        {
            DBG_LOG("刷新连接:%p超时任务", this);
            return _loop->RefrushTimerTask(_connection_id);
        }
        // DBG_LOG("添加连接:%p超时任务", this);
        _loop->TimerTaskAdd(_connection_id, delay, std::bind(&Connection::Realse, this));
    }
    void SendInLoop(Buffer &buf)
    {
        // 如果此时连接已经关闭 不要发送数据
        if (_connection_status == DISCONNECTED)
        {
            DBG_LOG("连接已断开，跳过发送: %p", this);
            return;
        }
        if (_sock_fd <= 0)
        {
            DBG_LOG("无效的文件描述符: %d", _sock_fd);
            return;
        }
        // 把数据发送到我们的发送缓冲区中
        _send_buffer.WriteBufferAndPush(buf);
        // 开启写事件
        if (_channel.IsWriteEvent() == false)
            _channel.EnableWrite();
    }
    void RealseInLoop()
    {
        if (_connection_status == DISCONNECTED)
        {
            DBG_LOG("连接已释放，跳过重复释放: %p", this);
            return;
        }
        DBG_LOG("开始释放连接: %p, fd: %d", this, _sock_fd);
        _connection_status = DISCONNECTED;

        if (_loop->HasTimer(_connection_id)) // 如果有定时器任务 关闭的时候取消取消 取消就是不断刷新定时任务活跃度就可以了
            CancelInactiveReleaseInLoop();

        if (_channel.GetEvent() != 0)
        {
            // 移出epoll监控
            _channel.Remove();
        }

        // 关闭连接
        _socket.Close();
        _sock_fd = -1;
        // 设置连接状态
        DBG_LOG("连接释放完成: %p, fd: %d", this, _sock_fd);

        //  调用关闭回调函数，避免先移除服务器管理的连接信息导致Connection被释放，再去处理会出错，因此先调用用户的回调函数
        if (_closed_callback)
            _closed_callback(shared_from_this());
        // 移除服务器内部管理的连接信息
        if (_server_closed_callback)
            _server_closed_callback(shared_from_this());
    }

    void ShutDownInLoop()
    {
        if (_connection_status == DISCONNECTED)
            return;
        if (_recv_buffer.ReadSpace() > 0)
        {
            if (_message_callback)
            {
                _message_callback(shared_from_this(), &_recv_buffer);
            }
        }
        // 下面设置状态
        _connection_status = DISCONNECTING;
        if (_send_buffer.ReadSpace() > 0)
        {
            // 有数据的时候 此时我们看一下可写事件是否就绪
            if (_channel.IsWriteEvent() == false)
            {
                // 开启写事件 写事件失败就会进行断开连接
                _channel.EnableWrite();
            }
        }
        if (_send_buffer.ReadSpace() == 0)
        {
            if (_channel.IsWriteEvent() == true)
                _channel.DisableWrite();
        }
        // 关闭连接
        Realse();
    }
    void UpgradeInLoop(const Any &context, const ConnectedCallback &conn, const MessageCallback &msg, const ClosedCallback &closed, const AnyEventCallback &event)
    {
        _context = context;
        _connected_callback = conn;
        _message_callback = msg;
        _closed_callback = closed;
        _event_callback = event;
    }

public:
    Connection(uint64_t id, int sockfd, EventLoop *loop)
        : _connection_id(id), // id也是定时销毁连接的id
          _connection_status(CONNECTING),
          _sock_fd(sockfd),
          _socket(_sock_fd),
          _loop(loop),
          _channel(_sock_fd, _loop),
          _connected_callback(nullptr),
          _message_callback(nullptr),
          _closed_callback(nullptr),
          _event_callback(nullptr),
          _server_closed_callback(nullptr),
          _is_enable_realase(false)
    {

        // 设置事件的回调函数
        _channel.SetReadCallback(std::bind(&Connection::HandlerRead, this));
        _channel.SetWriteCallback(std::bind(&Connection::HandlerWrite, this));
        _channel.SetCloseCallback(std::bind(&Connection::HandlerClose, this));
        _channel.SetAllEventCallback(std::bind(&Connection::HanlerEvent, this));
        _channel.SetErrorCallback(std::bind(&Connection::HandlerErr, this));
    }
    ~Connection()
    {
        DBG_LOG("conect realse:%p", this);
    }
    // 设置回调函数
    void SetConnectedCallback(const ConnectedCallback &cb) { _connected_callback = cb; }
    void SetMessageCallback(const MessageCallback &cb) { _message_callback = cb; }
    void SetClosedCallback(const ClosedCallback &cb) { _closed_callback = cb; }
    void SetAnyEventCallback(const AnyEventCallback &cb) { _event_callback = cb; }
    void SetSrvClosedCallback(const ClosedCallback &cb) { _server_closed_callback = cb; }
    // 再关闭连接要不缓存区的数据处理完成
    void ShutDown()
    {
        _loop->RunInLoop(std::bind(&Connection::ShutDownInLoop, this));
    }
    // 关闭连接 任何和关闭连接的操作 都要进行压入任务队列，等处理完本次事件循环后进行释放操作
    void Realse()
    {
        _loop->RunInLoop(std::bind(&Connection::RealseInLoop, this));
    }
    // 初始化连接进行开启我们的读监控 以及外界回调函数的设置
    void EnableConRead()
    {
        _loop->RunInLoop(std::bind(&Connection::EnableConReadInLoop, this));
    }
    int Id()
    {
        return _connection_id;
    }
    int Fd()
    {
        return _sock_fd;
    }
    Any *Context()
    {
        return &_context;
    }

    void SetContext(const Any &context)
    {
        _context = context;
    }
    // 开启不活跃连接的释放 添加定时任务
    void EnableInactiveRelse(int delay)
    {
        _loop->RunInLoop(std::bind(&Connection::EnableInactiveRelseInLoop, this, delay));
    }
    // 取消定时任务 就是进行活跃度的刷新
    void CancelInactiveRelease()
    {
        _loop->RunInLoop(std::bind(&Connection::CancelInactiveReleaseInLoop, this));
    }
    // 业务处理完把数据交给我的发送缓冲区
    void Send(const char *buffer, uint64_t len)
    {
        Buffer tem;
        tem.WriteAndPop(buffer, len);
        _loop->RunInLoop(std::bind(&Connection::SendInLoop, this, std::move(tem)));
    }
    // 切换我们的协议 就是切换上下文 以及我们外部的回调函数的设置
    void Upgrade(const Any &context, const ConnectedCallback &conn, const MessageCallback &msg, const ClosedCallback &closed, const AnyEventCallback &event)
    {
        // 注意我们要保证切换协议的要先执行 如果放在任务队列中 此时事件就绪 就会继续上次没切换的协议进行处理数据
        _loop->AssertInLoop();
        // 走到这里此时协议一定不会加入队列 会立刻执行
        _loop->RunInLoop(std::bind(&Connection::UpgradeInLoop, this, context, conn, msg, closed, event));
    }
};

class Acceptor
{
    using AcceptCallback = std::function<void(int)>;

private:
    Socket _listensock;
    EventLoop *_base_loop;
    channel _channel;
    AcceptCallback _accept_callback;

private:
    int CreateServe(uint16_t port)
    {
        _listensock.CreateServe(port);
        return _listensock.Fd();
    }
    void HandleAccept()
    {
        int fd = _listensock.Accept();
        if (fd < 0)
        {
            ERR_LOG("获取连接fd出错");
            return;
        }
        if (_accept_callback)
        {
            _accept_callback(fd);
        }
    }

public:
    /*不能将启动读事件监控，放到构造函数中，必须在设置回调函数后，再去启动*/
    /*否则有可能造成启动监控后，立即有事件，处理的时候，回调函数还没设置：新连接得不到处理，且资源泄漏*/
    Acceptor(uint16_t port, EventLoop *loop)
        : _listensock(CreateServe(port)),
          _base_loop(loop),
          _channel(_listensock.Fd(), _base_loop),
          _accept_callback(nullptr)

    {
        // 设置我们的读事件回调函数
        _channel.SetReadCallback(std::bind(&Acceptor::HandleAccept, this));
    }
    void SetAcceptCallBack(const AcceptCallback &cb)
    {
        _accept_callback = cb;
    }
    void listen()
    {
        // 开启读事件监控
        _channel.EnableRead();
    }
};
// 当我们线程函数执行时如果此时我们调用获取我们的eventloop可能会获得空指针
// 因此我们引入锁和条件变量 保持主线程和从线程的同步
class LoopThread
{
private:
    std::thread _thread;
    EventLoop *_loop;
    std::mutex _mutex;
    std::condition_variable _cond; // 条件变量
    // 线程入口函数 负责申请一个eventloop对象
    void EntryThread()
    {
        INF_LOG("初始线程id:%lu", pthread_self());
        //  局部变量 初始化后复制给成员变量
        EventLoop loop;
        {
            std::unique_lock<std::mutex> lock(_mutex);
            _loop = &loop;
            // 唤醒因为外界获取eventllop为空导致阻塞的线程
            _cond.notify_all();
        }
        // 进行事件循环
        _loop->Start();
    }

public:
    LoopThread()
        : _thread(std::bind(&LoopThread::EntryThread, this)),
          _loop(nullptr)
    {
    }
    // 我们在分配连接的时候需要拿到线程内的eventpoll
    EventLoop *GetEventLoop()
    {
        EventLoop *loop = nullptr;
        {
            std::unique_lock<std::mutex> _lock(_mutex);
            _cond.wait(_lock, [&]() -> bool
                       { return _loop != nullptr; });
            loop = _loop;
        }
        return loop;
    }
};

class LoopThreadPool
{
private:
    // 线程数量
    int _thredad_count;
    // 分配线程的线程下标
    int _thread_ind;
    // 我们的主线程
    EventLoop *_baseloop;
    // 线程池
    std::vector<LoopThread *> _threads;
    // eventloop的容器
    std::vector<EventLoop *> _loops;

public:
    LoopThreadPool(EventLoop *baseloop)
        : _thredad_count(0),
          _thread_ind(-1),
          _baseloop(baseloop)
    {
    }
    void SetCount(int count)
    {
        _thredad_count = count;
    }
    // 创建一批线程
    void Create()
    {
        _threads.resize(_thredad_count);
        _loops.resize(_thredad_count);
        for (int i = 0; i < _thredad_count; i++)
        {

            _threads[i] = new LoopThread();
            // 内部创建的子线程不创建出一个eventloop主线程就会一直阻塞
            // 直到子线程创建完毕后 可以进行获取
            _loops[i] = _threads[i]->GetEventLoop();
        }
    }
    EventLoop *NextEventLoop()
    {
        if (_thredad_count == 0)
        {
            return _baseloop;
        }
        _thread_ind = (_thread_ind + 1) % _thredad_count;
        return _loops[_thread_ind];
    }
};

class TcpServe
{
    using TimerFunc = std::function<void()>;
    using ConnectedCallback = std::function<void(const PtrConnection &)>;
    using MessageCallback = std::function<void(const PtrConnection &, Buffer *)>;
    using ClosedCallback = std::function<void(const PtrConnection &)>;
    using AnyEventCallback = std::function<void(const PtrConnection &)>;

private:
    int _next_id; // 作为连接id以及我们的定时任务id
    int _timeout;
    bool _enable_inactive_realease;
    EventLoop _baseloop;
    Acceptor _accept;
    LoopThreadPool _pool;
    std::unordered_map<int, PtrConnection> _conns;

    ConnectedCallback _connected_callback;
    MessageCallback _message_callback;
    ClosedCallback _closed_callback;
    AnyEventCallback _event_callback;
    ClosedCallback _server_closed_callback;

private:
    void ListenSocketReadcb(int newfd)
    {
        // 为通信套接字创建事件管理 并指定所属的epoll实例
        // channel commsock(fd, p); //这里的设置进去的连接不能是局部不然的话函数介绍变量销毁 此时回调中的channel为空
        // INF_LOG("baseloop:%lu", pthread_self());
        PtrConnection commsock(new Connection(_next_id, newfd, _pool.NextEventLoop()));
        DBG_LOG("创建新连接: %p, fd: %d, id: %d", commsock.get(), newfd, _next_id);
        _conns.insert(std::make_pair(_next_id++, commsock));
        // 设置读事件处理函数
        commsock->SetConnectedCallback(_connected_callback);
        commsock->SetMessageCallback(_message_callback);
        commsock->SetClosedCallback(_closed_callback);
        commsock->SetSrvClosedCallback(std::bind(&TcpServe::RemoveConnection, this, std::placeholders::_1));
        commsock->SetAnyEventCallback(_event_callback);
        // 我们创建线程的时候已经开启了每个eventloop的事件监控
        // 此时事件轮已经运作了
        // 如果我们先进行通信套接字的开启
        // 去处理读取失败的情况下
        // 会进行关闭连接的操作 此时就会取消超时销毁
        // 但是如果我们的超时销毁没有设置进去的话 就会导致超时任务没有取消
        if (_enable_inactive_realease)
        {

            commsock->EnableInactiveRelse(_timeout);
        }
        commsock->EnableConRead();
    }
    void RemoveConnectionInLoop(const PtrConnection &conc)
    {
        int id = conc->Id();
        auto it = _conns.find(id);
        if (it != _conns.end())
        {
            _conns.erase(it);
        }
    }
    void RemoveConnection(const PtrConnection &conc)
    {
        _baseloop.RunInLoop(std::bind(&TcpServe::RemoveConnectionInLoop, this, conc));
    }

    void AddTimerTaskInLoop(int delay, const TimerFunc &fuc)
    {
        _baseloop.TimerTaskAdd(_next_id, delay, fuc);
    }

public:
    TcpServe(int port)
        : _next_id(1),
          _enable_inactive_realease(false),
          _accept(port, &_baseloop),
          _pool(&_baseloop)
    {
        _accept.SetAcceptCallBack(std::bind(&TcpServe::ListenSocketReadcb, this, std::placeholders::_1));
        //_accept对象里面会自己创建服务器
        // 进行监控 连接到来获取文件描述符 此时调用accept回调函数进行连接分配
        _accept.listen();
    }

    void SetConnectedCallback(const ConnectedCallback &cb) { _connected_callback = cb; }
    void SetMessageCallback(const MessageCallback &cb) { _message_callback = cb; }
    void SetClosedCallback(const ClosedCallback &cb) { _closed_callback = cb; }
    void SetAnyEventCallback(const AnyEventCallback &cb) { _event_callback = cb; }
    // void SetSrvClosedCallback(const ClosedCallback &cb) { _server_closed_callback = cb; }
    void SetThreadCount(int count)
    {
        _pool.SetCount(count);
    }
    void Start()
    {
        _pool.Create();
        // 此时开启事件循环 在这里面开启之后才会有监听事件就绪进行处理
        _baseloop.Start();
    }
    // 添加定时任务必须在主线程中进行 因为我们要进行分配我们定时任务id
    void AddTimerTask(int delay, const TimerFunc &fuc)
    {
        _baseloop.RunInLoop(std::bind(&TcpServe::AddTimerTaskInLoop, this, delay, fuc));
    }
    void EnableInactiveRelase(int delay)
    {
        _timeout = delay;
        _enable_inactive_realease = true;
    }
};

void channel::Remove()
{
    // 需要我们的poller模块
    _poll->RemoveEvent(this);
}
// 把事件设置到内核中
void channel::IntoEpoll()
{
    _poll->UpdateEvent(this);
}
// 三个外部接口有可能会被其他线程操作加入定时任务 因此要确保线程安全让其操作在自己的线程中执行
void TimerWheel::TimerTaskAdd(int id, int timeout, TaskFunc func)
{
    // 这里如果是其他线程执行该函数的时候
    //_loop里面的线程id一定不是和当前线程id相等
    // 确保该任务是在我的loop线程被执行的
    _loop->RunInLoop(std::bind(&TimerWheel::TimerTaskAddInLoop, this, id, timeout, func));
}
void TimerWheel::CancelTask(int id)
{
    _loop->RunInLoop(std::bind(&TimerWheel::CancelTaskInLoop, this, id));
}
// 刷新定时任务

void TimerWheel::Refresh(int id)
{
    _loop->RunInLoop(std::bind(&TimerWheel::RefreshInLoop, this, id));
}
// 连接关闭如果继续发送就会发送sigpipe信号导致服务器崩溃
class NetWork
{
public:
    NetWork()
    {
        DBG_LOG("SIGPIPE INIT");
        signal(SIGPIPE, SIG_IGN);
    }
};
static NetWork nw;

#endif